package com.education.realtime.app.dwd;

import com.education.realtime.app.BaseSQLApp;
import com.education.realtime.common.Constant;
import com.education.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradePaymentApp extends BaseSQLApp {

    public static void main(String[] args) {
        new DwdTradePaymentApp().init(2004,"DwdTradePaymentApp");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"DwdTradePaymentApp");

        //过滤出支付表数据
        Table carTable = tEnv.sqlQuery("select" +
                " `data`['id'] id, " +
                " `data`['out_trade_no'] out_trade_no, " +
                " `data`['order_id'] order_id, " +
                " `data`['alipay_trade_no'] alipay_trade_no, " +
                " `data`['total_amount'] total_amount, " +
                " `data`['trade_body'] trade_body, " +
                " `data`['payment_type'] payment_type, " +
                " `data`['payment_status'] payment_status, " +
                " `data`['create_time'] create_time, " +
                " `data`['update_time'] update_time, " +
                " `data`['callback_content'] callback_content, " +
                " `data`['callback_time'] callback_time, " +
                " ts, " +
                " xid " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='payment_info'"
        );

        //把结果写入kafka
        tEnv.executeSql("create table dwd_trade_payment(" +
                " id string, " +
                " out_trade_no string, " +
                " order_id string, " +
                " alipay_trade_no string, " +
                " total_amount string, " +
                " trade_body string, " +
                " payment_type string, " +
                " payment_status string, " +
                " create_time string, " +
                " update_time string, " +
                " callback_content string, " +
                " callback_time string, " +
                " ts String, " +
                " xid String " +
                ")" + SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_PAY_ADD)
        );
        carTable.executeInsert("dwd_trade_payment");


    }
}
